﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Fabric;
using System.Fabric.Description;
using System.IO;
using System.Net.Sockets;
using System.Reflection;
using System.Text;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    public class FabricActorClient : ActorClient
    {
        private long? m_hashedPartitionKey;  // Windows Fabric's Int64-based partitioning scheme maps a String to an Int64, then we pass that to Windows Fabric.

        // Cache the most recent ResolvedServicePartition
        private ResolvedServicePartition m_resolvedServicePartition = null;

        //
        // Class constructor, with PLib hacks
        //
        static FabricActorClient()
        {
            PLibHacks.ConsoleWriteLine = new Action<String>(Console.WriteLine);
            PLibHacks.TraceWriteLine = new Action<String>(s => Trace.WriteLine(s));
        }

        //
        // Constructor
        //

        public FabricActorClient(Uri fabricUri, Uri actorUri, bool useGateway, String partitionName = null)
            : base(fabricUri, actorUri, useGateway, partitionName)
        {
            Contract.Requires(actorUri != null);
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorUri.ToString().Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");

            if (partitionName != null)
                m_hashedPartitionKey = FabricServerClient.GetHashedPartitionKey(partitionName);

            SetupActorClientAsync().Wait();
        }

        private FabricActorClient(Uri fabricUri, Uri actorUri, bool useGateway, long? hashedPartitionKey, String partitionName = null)
            : base(fabricUri, actorUri, useGateway, partitionName)
        {
            Contract.Requires(actorUri != null);
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorUri.ToString().Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");

            m_hashedPartitionKey = hashedPartitionKey;

            // SetupActorClient(); Don't do this.  Let the static async creator take care of it
        }

        public static async Task<FabricActorClient> CreateAsync(Uri fabricUri, Uri actorUri, bool useGateway, string partitionName = null)
        {
            long? hashedPartitionKey = null;
            if (partitionName != null)
                hashedPartitionKey = FabricServerClient.GetHashedPartitionKey(partitionName);

            FabricActorClient result = new FabricActorClient(fabricUri, actorUri, useGateway, hashedPartitionKey, partitionName);
            await result.SetupActorClientAsync().ConfigureAwait(continueOnCapturedContext: false);
            return result;
        }

        public FabricActorClient(Uri fabricUri, Uri actorUri, bool useGateway, Func<Uri, Uri, Uri> resolver)
            : base(fabricUri, actorUri, useGateway, null, resolver)
        {
            Contract.Requires(actorUri != null);
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorUri.ToString().Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");

            SetupActorClientAsync().Wait();
        }

        //
        // Factory methods
        //

        public static void CreateStatelessActor(string fabricAddress, string actorInstanceName, StatelessActorStateTypes stateType, StatelessActorPersonalities personality, int numPartitions=1)
        {
            Contract.Requires(!String.IsNullOrWhiteSpace(actorInstanceName));
            Contract.Requires(numPartitions >= 1);

            FabricClient fabricClient = FabricServerClient.GetFabricClient(fabricAddress);
            CreateStatelessActor(fabricClient, actorInstanceName, stateType, personality, numPartitions);
        }

        private static void CreateStatelessActor(FabricClient fabricClient, string actorInstanceName, StatelessActorStateTypes stateType, StatelessActorPersonalities personality, int numPartitions)
        {
            Contract.Requires(fabricClient != null);
            Contract.Requires(!String.IsNullOrWhiteSpace(actorInstanceName));

            string personalityString;
            switch (personality)
            {
                case StatelessActorPersonalities.Dictionary: personalityString = "Dictionary"; break;
                case StatelessActorPersonalities.Empty: personalityString = "Empty"; break;
                case StatelessActorPersonalities.List: personalityString = "List"; break;
                default: throw new InvalidOperationException("Unknown personality " + personality);
            }

            string stateTypeString;
            switch (stateType)
            {
                case StatelessActorStateTypes.InMemory: stateTypeString = "InMemory"; break;
                default: throw new InvalidOperationException("Unknown state type " + stateType);
            }

            var partitionScheme = GetPartitionSchemeDescription(numPartitions);
            var statelessDescription = new StatelessServiceDescription();
            statelessDescription.ServiceName = new Uri(actorInstanceName);
            statelessDescription.ServiceTypeName = "StatelessActorService";
            statelessDescription.PartitionSchemeDescription = partitionScheme;
            statelessDescription.ApplicationName = new Uri("fabric:/actor/stateless");
            var initializationInfo = String.Format("StateType={0};Personality={1}", stateTypeString, personalityString);
            statelessDescription.InitializationData = Encoding.UTF8.GetBytes(initializationInfo);

            // Create a service/actor, flowing exceptions to the caller
            try
            {
                fabricClient.ServiceManager.CreateServiceAsync(statelessDescription).Wait();
            }
            catch (FabricElementAlreadyExistsException)
            {
                // Swallow these.  
            }
            catch (AggregateException ae)
            {
                // Ignore if this stems from FabricElementAlreadyExistsException
                if (!(ae.GetBaseException() is FabricElementAlreadyExistsException))
                    throw;
            }

        }

        // Static factory method for creating actors/services
        public static void CreateActor(string fabricAddress, string applicationType, string actorInstanceName, string serviceType, int numReplicas = 2, int numPartitions = 1)
        {
            Contract.Requires(!String.IsNullOrWhiteSpace(applicationType));
            Contract.Requires(!String.IsNullOrWhiteSpace(actorInstanceName));
            Contract.Requires(!String.IsNullOrWhiteSpace(serviceType));
            Contract.Requires(numReplicas >= 1, "Number of replicas must be one or more.");
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");

            FabricClient fabricClient = FabricServerClient.GetFabricClient(fabricAddress);

            CreateActor(fabricClient, applicationType, actorInstanceName, serviceType, numReplicas, numPartitions);
        }

        private static void CreateActor(FabricClient fabricClient, string applicationType, string actorInstanceName, 
            string serviceType, int numReplicas, int numPartitions)
        {
            Contract.Requires(fabricClient != null);
            Contract.Requires(!String.IsNullOrWhiteSpace(applicationType));
            Contract.Requires(!String.IsNullOrWhiteSpace(actorInstanceName));
            Contract.Requires(!String.IsNullOrWhiteSpace(serviceType));
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorInstanceName.Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");
            Contract.Requires(numReplicas >= 1, "Number of replicas must be one or more.");
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");

            if (numReplicas <= 0)
                throw new ArgumentOutOfRangeException("numReplicas", "Number of replicas must be one or more.");

            PartitionSchemeDescription partitionScheme = GetPartitionSchemeDescription(numPartitions);
            StatefulServiceDescription statefulDescription = new StatefulServiceDescription();
            statefulDescription.ServiceName = new Uri(actorInstanceName);
            statefulDescription.ServiceTypeName = serviceType;
            statefulDescription.PartitionSchemeDescription = partitionScheme;
            statefulDescription.TargetReplicaSetSize = numReplicas;
            statefulDescription.HasPersistedState = false;
            statefulDescription.MinReplicaSetSize = ((numReplicas % 2) == 0) ? (numReplicas + 1) / 2 : (numReplicas / 2) + 1;
            statefulDescription.ApplicationName = new Uri(applicationType);

            // Create a service/actor, flowing exceptions to the caller.
            // CreateServiceAsync uses a default timeout which as of v2 was 1 minute.  We can hit in Azure under significant load.  
            // We could lengthen that timeout, or try again.
            try
            {
                fabricClient.ServiceManager.CreateServiceAsync(statefulDescription).Wait();
            }
            catch (FabricElementAlreadyExistsException)
            {
                // Swallow these.  
            }
            catch (AggregateException ae)
            {
                // Ignore if this stems from FabricElementAlreadyExistsException
                if (!(ae.GetBaseException() is FabricElementAlreadyExistsException))
                    throw;
            }
        }

        // Translates an actor-level concept to Windows Fabric types.
        private static PartitionSchemeDescription GetPartitionSchemeDescription(int numPartitions)
        {
            Contract.Requires(numPartitions >= 1);
            Contract.Ensures(Contract.Result<PartitionSchemeDescription>() != null);

            if (numPartitions == 1)
                return new SingletonPartitionSchemeDescription();

            PartitionSchemeDescription partitionScheme = new UniformInt64RangePartitionSchemeDescription(numPartitions, Int32.MinValue, Int32.MaxValue);
            return partitionScheme;
        }

        /// <summary>
        /// Create an actor if it doesn't exist, or connect to it if it does.  Then, return a FabricActorClient.
        /// </summary>
        /// <param name="fabricAddress">Address and port of cluster to connect to, such as 
        /// MyApp.cloudapp.net:19000, localhost:19000, or 127.0.0.1:19000</param>
        /// <param name="applicationType">Type of application to create, such as "fabric:/actor/list"</param>
        /// <param name="actorUri">Address of the specific actor in the cluster, such as "fabric:/actor/list/MyList"</param>
        /// <param name="actorTypeName">A class name used for each individual replica, such as "ListActorReplica"</param>
        /// <param name="connectThroughGateway">Whether to connect using an Azure-aware gateway to route requests.</param>
        /// <param name="numReplicas">Number of instances of your service in the cloud.  Must be at least 1.</param>
        /// <param name="numPartitions">Number of partitions for service.  Must be at least 1.</param>
        /// <param name="partitionName">If the number of partitions is greater than 1, then this defines a name 
        /// that maps to a particular partition within the service.</param>
        /// <returns>A new FabricActorClient instance talking to the right actor.</returns>
        public new static FabricActorClient CreateOrConnectToActor(string fabricAddress, string applicationType, Uri actorUri, 
            string actorTypeName, bool connectThroughGateway, int numReplicas = 2, 
            int numPartitions = 1, String partitionName = null)
        {
            Contract.Requires(actorUri != null);
            Contract.Requires(Contract.ForAll(ReservedDelimiters, reservedChar => !actorUri.ToString().Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");
            Contract.Requires(numReplicas >= 1, "Number of replicas must be one or more.");
            Contract.Requires(numPartitions >= 1, "Actors must have one or more partitions.");
            Contract.Requires(numPartitions == 1 || partitionName != null, 
                "Using multiple partitions requires a partition name for the individual partition you want to connect to.");

            // See if this service instance exists.  If not, create it.
            Trace.WriteLine("CreateOrConnectToActor(" + actorUri + (partitionName == null ? "" : ", " + partitionName) + ")", "Information");
            FabricClient fabricClient = FabricServerClient.GetFabricClient(fabricAddress);
            bool serviceExists = false;
            const int numNameExistsRetries = 3;
            for (int i = 0; i < numNameExistsRetries; i++)
            {
                try
                {
                    // Saw timeouts when connecting with Azure, under significant load.  Took about 2 minutes to time out?
                    serviceExists = fabricClient.PropertyManager.NameExistsAsync(actorUri).Result;
                }
                catch (TimeoutException)
                {
                    if (i == numNameExistsRetries - 1)
                        throw;
                }
            }

            if (!serviceExists)
            {
                // BUGBUG: Race condition exists here.  Contact Actor Runtime experts for a better written method.
                try
                {
                    FabricActorClient.CreateActor(fabricClient, applicationType, actorUri.ToString(), actorTypeName, numReplicas, numPartitions);
                }
                catch (AggregateException agg)
                {
                    if (agg.GetBaseException() is FabricElementNotFoundException)
                    {
                        Contract.Assert(false, String.Format("Actor of type {0} was not deployed to the Actor Runtime.  Detailed error message: {1}", actorTypeName, agg.GetBaseException().Message));
                    }
                    throw;
                }
            }
            Uri fabricUri = new Uri(String.Format("net.tcp://{0}", fabricAddress));
            return new FabricActorClient(fabricUri, actorUri, connectThroughGateway, partitionName);
        }

        public static void DeleteActor(string fabricAddress, string actorInstanceName, bool deleteAllPartitions = false)
        {
            Contract.Requires(!String.IsNullOrEmpty(actorInstanceName));

            DeleteActor(fabricAddress, actorInstanceName, null, deleteAllPartitions);
        }

        private static void DeleteActor(string fabricAddress, string actorInstanceName, ActorClient actorClient, bool deleteAllPartitions)
        {
            Contract.Requires(!String.IsNullOrEmpty(actorInstanceName));

            // If we have an ActorClient to this actor, send a message to the actor client to delete itself.
            // TODO: Think about how to allocate a FabricActorClient (with a few more parameters) to send a delete message.
            // Perhaps this should be an instance method only...
            if (actorClient != null)
            {
                Contract.Assert(!deleteAllPartitions || actorClient.PartitionName != null, "Deleting all partitions only works on a partitioned actor.");

                try {
                    Task t = actorClient.SendDeleteRequest();
                    t.Wait();
                }
                catch (Exception) { }

                actorClient.Dispose();
            }

            // If this is a partitioned actor and we did not ask to delete all partitions, then
            // return.  But, if we may not have an ActorClient or any convenient way of detecting whether
            // we are dealing with a partitioned actor.
            if (actorClient != null && actorClient.PartitionName != null && !deleteAllPartitions)
                return;

            FabricClient fabricClient = FabricServerClient.GetFabricClient(fabricAddress);

            try
            {
                Task t = fabricClient.ServiceManager.DeleteServiceAsync(new Uri(actorInstanceName));
                t.Wait();
            }
            catch (FabricException)
            {
                // Swallow this.  It probably means that the actor was not found.
            }
            catch (AggregateException ag)
            {
                // Swallow FabricElementNotFoundExceptions
                if (ag.InnerExceptions.Count != 1 || ag.InnerException.GetType() != typeof(FabricElementNotFoundException))
                    throw;
            }
        }

        public override void DeleteActor()
        {
            String fabricAddress = m_serverUri.Host + ":" + m_serverUri.Port;
            String actorInstanceName = m_actorUri.ToString();
            DeleteActor(fabricAddress, actorInstanceName, this, false);
        }

        // Used by an interesting relay.
        public ResolvedServicePartition WorkingServicePartition
        {
            get { return m_resolvedServicePartition; }
        }

        //
        // Wrappers for IActor methods
        //

        #region Assembly Deployment

        public void DeployAssemblyIfNeeded(Assembly assembly, bool replaceAllVersions = false, String pathToX64CompatibleBinary = null)
        {
            Contract.Requires(assembly != null);
            String fullName = assembly.FullName;
            if (!IsFrameworkOrOSAssembly(assembly) && !DoesAssemblyExist(fullName, !replaceAllVersions))
            {
                // Must make sure that the x64 or MSIL flavor of the assembly is deployed.  We have the currently running version (ARM, x86, x64).
                String pathToAssembly = pathToX64CompatibleBinary ?? assembly.Location;
                if (pathToX64CompatibleBinary == null)
                {
                    ProcessorArchitecture asmArch = assembly.GetName().ProcessorArchitecture;
                    if (asmArch != ProcessorArchitecture.MSIL && asmArch != ProcessorArchitecture.Amd64)
                        throw new BadImageFormatException(String.Format("Actors currently only support running on x64, and need either MSIL or x64 assemblies.  Assembly {0}'s architecture of {1} is not supported.", assembly.GetName().Name, asmArch), assembly.Location);
                }
                byte[] bytes = File.ReadAllBytes(pathToAssembly);
                AddAssembly(fullName, bytes/*, replaceAllVersions*/);
            }
        }

        private const String ClrEcmaPublicKeyToken = "b77a5c561934e089";
        private const String DotNetFrameworkKeyToken = "b03f5f7f11d50a3a";
        private const String WindowsVistaKeyToken = "31bf3866ad364e35";

        private static bool IsFrameworkOrOSAssembly(Assembly assembly)
        {
            byte[] publicKeyToken = assembly.GetName().GetPublicKeyToken();
            if (publicKeyToken.Length == 0)
                return false;
            String token = ByteArrayToHexString(publicKeyToken);
            return (token.Equals(ClrEcmaPublicKeyToken) || token.Equals(DotNetFrameworkKeyToken) || token.Equals(WindowsVistaKeyToken));
        }

        private static String ByteArrayToHexString(byte[] key)
        {
            if (key.Length == 0)
                return String.Empty;
            StringBuilder sb = new StringBuilder(key.Length * 2);
            foreach (byte b in key)
                sb.Append(b.ToString("x2"));
            return sb.ToString();
        }

        #endregion Assembly Deployment

        //
        // Private utility methods and classes
        //

        #region Channel setup
        // Initializes the actor proxy (m_actorClient)
        // Called from the constructor, and from any logic that re-resolves the actor.
        // Note: sometimes we get back results that say deterministically, give up.  We'll return false.
        protected override async Task<bool> SetupActorClientAsync(bool registerFaultedCallback = true)
        {
            m_resolveState = ResolverState.Resolving;
            if (m_resolver != null)
            {
                // using custom resolver. Get endpoint Uri
                Uri uri = m_resolver(m_useGateway ? m_gatewayUri : m_serverUri, m_actorUri);

                // create Client to resolved address
                var actorClient = new WrappedTcpClient(uri.Host, uri.Port);
                m_actorTcpClient = await ActorRawTcpClient.CreateAsync(actorClient, this, m_useGateway ? m_actorUri.ToString() : uri.LocalPath, m_partitionName)
                    .ConfigureAwait(continueOnCapturedContext:false);
            }
            else
            {
                if (m_useGateway)
                {
                    m_actorTcpClient = await AzureClientSetupAsync(m_gatewayUri, m_actorUri).ConfigureAwait(continueOnCapturedContext: false);
                }
                else
                {
                    m_actorTcpClient = await ServerClientSetupAsync(m_serverUri, m_actorUri).ConfigureAwait(continueOnCapturedContext: false);
                    //Console.WriteLine("SetupActorClient: connected to remote endpoint {0}", m_actorClient.RemoteEndpoint);
                }
            }

            if (m_actorTcpClient == null)
            {
                m_resolveState = ResolverState.Resolved;  // Resolved that it's not here.
                return false;
            }

            // If requested, register for a callback when the channel faults
            if (registerFaultedCallback)
            {
                // Receive (proactive) notification if the channel faults
                m_actorTcpClient.Faulted += new EventHandler(ActorClient_Faulted);
            }

            m_resolveState = ResolverState.Resolved;
            return true;
        }


        // Returns an actor proxy for an actor/service on a server (most probably the local machine)
        private async Task<ActorRawTcpClient> ServerClientSetupAsync(Uri fabricUri, Uri actorUri)
        {
            FabricServerClient serverClient;

            ResolvedServicePartition currentResolvedServicePartition;
            TcpClient tcpClient;
            Exception lastException = null;

            // Consider a service that gets load balanced.  Imperically, we've seen this take about 16 seconds from the 
            // time that the service starts load balancing til the point where the service is ready again.  This is with
            // moderate load as well, on a five node dev machine.  Notably, we can resolve the service in the mean time,
            // but won't be able to connect to it because no one is listening on a particular port!  So we have a number
            // of failures that we must retry repeatedly.
            for (int connectRetry = 0; connectRetry < CONNECT_RETRIES; connectRetry++)
            {
                if (m_disposed)
                    return null;

                if (connectRetry > 0)
                {
                    // Ensure that this loop and this time limit work out to at least 16 seconds, probably about 30 seconds.
                    await Task.Delay(CONNECT_RETRY_WAIT_TIME_MS).ConfigureAwait(continueOnCapturedContext: false);
                    //Trace.WriteLine("FabricActorClient.ServerClientSetup: Failed to resolve " + actorUri.ToString() + ", partition name = \"" + m_partitionName + "\", trying again", "Warning");
                }

                // Creating a new FabricServerClient for each retry is more reliable than reusing the same one.
                if (fabricUri != null)
                {
                    var connectionAddress = String.Format("{0}:{1}", fabricUri.Host, fabricUri.Port);
                    serverClient = new FabricServerClient(new string[] { connectionAddress });
                }
                else serverClient = new FabricServerClient(new string[0]);

                string uniquePathInfo;

                var tuple = await serverClient.TryResolveAndCreateRawClientAsync(actorUri, m_hashedPartitionKey, m_resolvedServicePartition).ConfigureAwait(continueOnCapturedContext: false);
                if (tuple.Item2 != null)
                {
                    // Even if resolution failed (or we looked up a new location only to find it is incorrect),
                    // let's keep that resolved service partition around.  Hopefully this will help future resolution
                    // attempts to realize that its internal cache is out of date and must be refreshed.  This is recommended,
                    // but as of October 2013 appears to be insufficient to address stress bugs in resolution.
                    currentResolvedServicePartition = tuple.Item2;
                }
                
                if (tuple.Item4 == FabricServerClient.FabricResolveReturnCode.Success)
                {
                    tcpClient = tuple.Item1;
                    currentResolvedServicePartition = tuple.Item2;
                    uniquePathInfo = tuple.Item3;
                    m_resolvedServicePartition = currentResolvedServicePartition; // re-cache
                    try
                    {
                        var candidate = await ActorRawTcpClient.CreateAsync(new WrappedTcpClient(tcpClient), this, uniquePathInfo, m_partitionName).ConfigureAwait(continueOnCapturedContext: false);
                        if (await TestConnectivityAsync(candidate)) return candidate;
                        else
                        {
                            candidate.Dispose();
                            //Console.WriteLine("FAC::ServerClientSetupAsync - TestConnectivityFailed.  Attempt {0} of {1}", connectRetry, CONNECT_RETRIES);
                        }
                    }
                    catch (ActorNotFoundException anfe)
                    {
                        if (connectRetry > 0)
                        {
                            // My intent was to dump any cached values from the Actor Runtime and re-resolve.
                            // However the experts suggest that this is overkill and will only increase load on the system.
                            // Instead, pass in the previously-resolved service instance and let them update their caches.
                            //serverClient.CachedValuesAreInvalid();
                        }
                        lastException = anfe;
                        continue;  // Retry (including re-resolve)
                    }
                    catch (IOException ioe)
                    {
                        lastException = ioe;
                        continue;  // Retry (including re-resolve)
                    }
                    catch (SocketException se)
                    {
                        lastException = se;
                        continue;  // Retry (including re-resolve)
                    }
                    catch (AggregateException ae)
                    {
                        lastException = ae;
                        continue;
                    }
                    catch (InvalidOperationException invop)
                    {
                        lastException = invop;
                        // This can happen if the socket gets disconnected.
                        continue;
                    }
                    catch (Exception e)
                    {
                        lastException = e;
                        // What should we do here?
                        Console.WriteLine("FabricActorClient::ServerClientSetup got an unexpected exception.  " + e);
                        Debug.Assert(false, String.Format("Unexpected exception while creating an ActorRawTcpClient.  Retrying...\r\nError was {0}: {1}", e.GetType().Name, e.Message));
                        continue;
                    }
                }
                else if (tuple.Item4 == FabricServerClient.FabricResolveReturnCode.Retriable)
                {
                    if (m_disposed)
                        return null;
                    String actorName = (m_partitionName == null) ? actorUri.ToString() : actorUri.ToString() + ", " + m_partitionName;
                    Console.WriteLine("Actor {0} was temporarily unavailable.  Attempt {1} of {2}", actorName, connectRetry, CONNECT_RETRIES);
                    continue;
                }
                else
                {
                    if (m_disposed)
                        return null;
                    return null;
                    //Console.WriteLine("FAC::ServerClientSetupAsync - TryResolveAndCreateRawClientAsync failed, returning {0} for actor {1}.  Attempt {2} of {3}",
                    //    tuple.Item4, actorUri.ToString(), connectRetry, CONNECT_RETRIES);
                }
            }

            String message = String.Format("ServerClientSetup: FAILED to resolve actor name {0}  partition name \"{1}\"", actorUri.ToString(), m_partitionName);
            if (lastException != null)
            {
                lastException.Data["FabricActorClient Details"] = message;
                throw lastException;
            }
            else
                throw new Exception(message);
        }

        // Returns an IActor proxy for an actor/service running on Azure
        private async Task<ActorRawTcpClient> AzureClientSetupAsync(Uri gatewayAddress, Uri serviceName)
        {
            Exception resultException = null;
            for (int i = 0; i < CONNECT_RETRIES; i++)
            {
                try
                {
                    var tcpClient = new TcpClient(gatewayAddress.Host, gatewayAddress.Port);
                    var candidate = await ActorRawTcpClient.CreateAsync(new WrappedTcpClient(tcpClient), this, serviceName.ToString(), m_partitionName)
                        .ConfigureAwait(continueOnCapturedContext: false);
                    if (await TestConnectivityAsync(candidate)) return candidate;
                    candidate.Dispose();
                }
                catch (Exception e)
                {
                    Console.WriteLine("Exception initializing azure client({0},{1}): {2}.  Retrying",
                        gatewayAddress.ToString(), serviceName.ToString(), e.Message);
                    resultException = e;
                }

                await Task.Delay(CONNECT_RETRY_WAIT_TIME_MS).ConfigureAwait(continueOnCapturedContext:false); // Wait a bit before retrying
            }

            // If we don't succeed after a number of retries, then throw the most recent failure
            throw resultException;
        }

        #endregion
    }
}
